package cn.me.alphamq.broker.netty.nameserver;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.store.VersionHolder;
import cn.me.alphamq.common.constant.GlobalConstant;
import cn.me.alphamq.common.entity.req.ServiceHeartBeatReq;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.msg.TcpMsg;
import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 定时发送心跳包的任务的启动者
 *
 * @author f
 */
public class HeartBeatTaskRunner {

    private static final Logger log = LoggerFactory.getLogger(HeartBeatTaskRunner.class);

    /**
     * 保证任务只被启动一次的标记
     */
    private final AtomicInteger flag = new AtomicInteger(0);

    /**
     * 保证任务只被启动一次，做了消费幂等的处理
     */
    public void startTask() {
        if (flag.getAndIncrement() == 0) {
            log.info("success register to nameserver...");
            new Thread(new HeartBeatTask(), "heart-beat-task").start();
        }
    }

    // 定时发送心跳包的任务类
    private static class HeartBeatTask implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    // 休眠
                    TimeUnit.MILLISECONDS.sleep(GlobalConstant.SERVICE_HEART_BEAT_TIME_STEP);

                    // 发送心跳包，携带 topic 和消费偏移量的信息
                    Container.getNameServerClient().getChannel().writeAndFlush(new TcpMsg(
                            EventCodeEnum.HEART_BEAT_REQ,
                            JSON.toJSONBytes(new ServiceHeartBeatReq(
                                    VersionHolder.getVersion(),
                                    Container.getCommitLogTopicModelList(),
                                    Container.getConsumeQueueOffsetModel()
                            ))
                    ));
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}
